package org.vison.wonfu.framework.soa.protocol;

import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.SneakyThrows;
import org.apache.hc.client5.http.async.methods.SimpleHttpRequest;
import org.apache.hc.client5.http.async.methods.SimpleHttpResponse;
import org.apache.hc.client5.http.config.RequestConfig;
import org.apache.hc.client5.http.impl.async.CloseableHttpAsyncClient;
import org.apache.hc.client5.http.impl.async.HttpAsyncClients;
import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManager;
import org.apache.hc.client5.http.impl.nio.PoolingAsyncClientConnectionManagerBuilder;
import org.apache.hc.core5.concurrent.DefaultThreadFactory;
import org.apache.hc.core5.concurrent.FutureCallback;
import org.apache.hc.core5.pool.PoolConcurrencyPolicy;
import org.apache.hc.core5.pool.PoolReusePolicy;
import org.apache.hc.core5.util.TimeValue;
import org.apache.hc.core5.util.Timeout;
import org.springframework.context.ApplicationContext;
import org.springframework.core.env.Environment;
import org.vison.wonfu.framework.commons.Constants;
import org.vison.wonfu.framework.commons.tools.Assert;
import org.vison.wonfu.framework.commons.utils.StringUtils;
import org.vison.wonfu.framework.commons.utils.TypeUtils;
import org.vison.wonfu.framework.soa.Invocation;
import org.vison.wonfu.framework.soa.SoaConstants;
import org.vison.wonfu.framework.soa.context.RpcContext;
import org.vison.wonfu.framework.soa.exception.RpcException;
import org.vison.wonfu.framework.soa.SoaPropertys;
import org.vison.wonfu.framework.soa.annotation.*;
import org.vison.wonfu.framework.soa.discovery.ServiceInstance;
import org.vison.wonfu.framework.soa.discovery.ServiceRegistryDiscovery;
import org.vison.wonfu.framework.soa.exception.SoaFallbackException;
import org.vison.wonfu.framework.soa.governance.MetaData;
import org.vison.wonfu.framework.soa.governance.SoaMetric;
import org.vison.wonfu.framework.soa.governance.circuitBreaker.CircuitBreakerManager;
import org.vison.wonfu.framework.soa.governance.circuitBreaker.Fallback;

import java.lang.reflect.Field;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.charset.StandardCharsets;
import java.security.KeyManagementException;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.util.*;
import java.util.concurrent.Future;

public class SoaRefProxyInvoke implements InvocationHandler, SoaRefInvocationHandler {
    private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(SoaRefProxyInvoke.class);

    private static CloseableHttpAsyncClient httpClient = createHighPerformanceClient();

    private static CloseableHttpAsyncClient createHighPerformanceClient() {
        try {
            // 1. 配置连接池
            PoolingAsyncClientConnectionManager connectionManager = configureConnectionManager();
            // 2. 配置请求级参数（超时、重试等）
            RequestConfig defaultRequestConfig = configureRequestConfig();
            // 3. 构建高性能客户端
            CloseableHttpAsyncClient client = HttpAsyncClients.custom()
                    .setConnectionManager(connectionManager)
                    .setDefaultRequestConfig(defaultRequestConfig)
                    .setThreadFactory(new DefaultThreadFactory("http-async-client", true)) // 命名线程池，便于监控
                    .evictExpiredConnections() // 定期清理过期连接
                    .evictIdleConnections(TimeValue.ofMinutes(5)) // 清理5分钟以上的空闲连接
                    .build();

            // 启动客户端
            client.start();
            return client;
        } catch (Exception e) {
            throw new RuntimeException("Failed to create high-performance HTTP client", e);
        }
    }

    private static PoolingAsyncClientConnectionManager configureConnectionManager()
            throws KeyStoreException, NoSuchAlgorithmException, KeyManagementException {

//        // 配置SSL上下文（支持TLSv1.2+，提升安全性和兼容性）
//        SSLContext sslContext = SSLContexts.custom()
//                .loadTrustMaterial((chain, authType) -> true) // 生产环境需替换为实际信任策略
//                .build();
//        TlsStrategy tlsStrategy = new DefaultClientTlsStrategy(sslContext);

        // 连接池配置
        return PoolingAsyncClientConnectionManagerBuilder.create()
//                .setTlsStrategy(tlsStrategy)
                .setPoolConcurrencyPolicy(PoolConcurrencyPolicy.STRICT) // 严格控制并发，避免连接泄露
                .setConnPoolPolicy(PoolReusePolicy.LIFO) // LIFO策略提升连接复用率
                .setMaxConnTotal(500) // 全局最大连接数（根据CPU核心数和服务能力调整）
                .setMaxConnPerRoute(100) // 单路由最大连接数（如单个域名）
                // 针对特定域名的连接数优化（如频繁访问的服务）
                // 连接配置
                .build();
    }

    private static RequestConfig configureRequestConfig() {
        return RequestConfig.custom()
                .setConnectionRequestTimeout(Timeout.ofSeconds(5)) // 从连接池获取连接的超时时间
                .setResponseTimeout(Timeout.ofSeconds(15)) // 整体响应超时（包含连接+处理+数据传输）
                .setRedirectsEnabled(false) // 禁用自动重定向（高性能场景通常手动处理）
                .setCircularRedirectsAllowed(false)
                .build();
    }


    static {
        ServiceLoader<ParamsHandler> paramsHandlerServiceLoader = ServiceLoader.load(ParamsHandler.class);
        paramsHandlerServiceLoader.iterator().forEachRemaining(paramsHandler -> {
            InvocationParamsHandler annotation = paramsHandler.getClass().getAnnotation(InvocationParamsHandler.class);
            if (annotation == null) {
                return;
            }
            ParamsHandlerFactory.register(annotation.protocol(), paramsHandler);
        });
    }


    private ApplicationContext applicationContext;

    private Field field;
    private ServiceRegistryDiscovery serviceRegistryDiscovery;

    private Environment environment;

    public SoaRefProxyInvoke(ApplicationContext applicationContext, Field field) {
        this.applicationContext = applicationContext;
        this.field = field;
        this.environment = applicationContext.getBean(Environment.class);
        this.serviceRegistryDiscovery = applicationContext.getBean(ServiceRegistryDiscovery.class);
    }

    /**
     * @param proxy  动态生成的代理对象本身（注意：通常不建议在 invoke 中直接使用该对象，避免循环调用）。
     * @param method 被调用的目标方法（Method 对象）。
     * @param args   调用目标方法时传入的参数数组。
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        logger.trace("Invoking SOA method: {}", method.getName());
        SoaRef soaRef = field.getAnnotation(SoaRef.class);
        Invocation invocation = buildInvocation(method, soaRef, args);
        // 从 spring boot 容器获取 ObjectMapper 实例
        ObjectMapper objectMapper = applicationContext.getBean(ObjectMapper.class);
        return doInvokeConsumer(invocation);
    }

    /**
     * 配置熔断器参数，例如超时时间、最大失败次数等
     *
     * @param key
     * @param breaker
     */
    private void configureCircuitBreaker(String key, CircuitBreakerManager.CircuitBreaker breaker) {
        SoaPropertys soaPropertys = applicationContext.getBean(SoaPropertys.class);
        breaker.setManualBreaker(soaPropertys.isConsumerMethodManualBreakerOpen(key));
        breaker.setEnable(soaPropertys.isConsumerMethodBreakerEnabled(key));
        breaker.setSlidingWindowSize(soaPropertys.getConsumerMethodSlidingWindowSize(key));
        breaker.setSuccessThreshold(soaPropertys.getConsumerMethodSuccessThreshold(key));
        breaker.setHalfOpenInterval(soaPropertys.getConsumerMethodHalfOpenInterval(key));
        breaker.setRequestThreshold(soaPropertys.getConsumerMethodRequestThreshold(key));
        breaker.setErrorRateThreshold(soaPropertys.getConsumerMethodErrorRateThreshold(key));
    }

    public Object doInvokeConsumer(Invocation invocation) throws Throwable {
        Protocol protocol = invocation.getProtocol();
        ParamsHandler paramsHandler = ParamsHandlerFactory.getParamsHandler(protocol);
        String circuitKey = invocation.getCommandKey();
        CircuitBreakerManager.CircuitBreaker breaker = CircuitBreakerManager.getOrCreate(circuitKey);
        configureCircuitBreaker(circuitKey, breaker);
        // 熔断检查
        if (!breaker.allowRequest()) {
            logger.warn("Circuit breaker is OPEN for method: {}", invocation.getMethodName());
            return handleFallback(invocation);
        }

        // 构建请求并获取请求头信息
        SimpleHttpRequest request = paramsHandler.pack(invocation, null);
        // 记录请求开始时间
        long startTime = System.currentTimeMillis();
        SimpleHttpResponse response = null;
        RpcException rpcException = null;
        long requestDuration = 0;
        try {
            Future<SimpleHttpResponse> future = httpClient.execute(request, new FutureCallback<SimpleHttpResponse>() {
                @Override
                public void completed(SimpleHttpResponse response) {
                    breaker.recordSuccess();
                }

                @Override
                public void failed(Exception ex) {
                    logger.error("SOA call failed with exception", ex);
                    breaker.recordFailure();
                }

                @Override
                public void cancelled() {
                    // 取消处理
                }
            });
            response = future.get();
            long endTime = System.currentTimeMillis();
            requestDuration = endTime - startTime;
            logConsumerRequest(invocation, request, response, requestDuration, rpcException);
            SoaMetric.recordRequestDuration(invocation, response, requestDuration);
            Class<?> returnType = invocation.getReturnType();
            if (returnType != null) {
                // 处理返回结果
                Object result = paramsHandler.unpack(response, invocation, null);
                return result;
            }
        }catch (RpcException e){
            rpcException = e;
            logConsumerRequest(invocation, request, response, requestDuration, rpcException);
        }finally {
            RpcContext.removeContext();
        }

        // 获取响应并处理

        return null;
    }


    @SneakyThrows
    private void logConsumerRequest(Invocation invocation, SimpleHttpRequest simpleHttpRequest,
                                    SimpleHttpResponse simpleHttpResponse, long requestDuration,RpcException rpcException) {
        String uri = simpleHttpRequest.getUri().toString();
        String requestParams = "";
        if (simpleHttpRequest.getBodyBytes() != null) {
            requestParams = new String(simpleHttpRequest.getBodyBytes());
        }
        StringBuilder headersBuilder = new StringBuilder();
        Arrays.stream(simpleHttpRequest.getHeaders()).forEach(header -> {
            headersBuilder.append(header.getName()).append(": ").append(header.getValue()).append(", ");
        });
        String requestHeaders = headersBuilder.length() > 0 ?
                headersBuilder.substring(0, headersBuilder.length() - 1) : "";
        if(rpcException != null){
            String msg = String.format("SOA调用 - 请求方法:%s, CommandKey:%s, URI:%s, 请求参数:%s, 请求头:%s, " +
                            "异常:%s",
                    invocation.getSoaRequestMethod(), invocation.getCommandKey(), uri, requestParams,
                    requestHeaders, rpcException.getMessage());
            logger.error(msg, rpcException);
            return;
        }
        byte[] bodyBytes = simpleHttpResponse.getBodyBytes();
        String responseBody = bodyBytes != null ? new String(bodyBytes, StandardCharsets.UTF_8) : "";
        String msg = String.format("SOA调用 - 请求方法:%s, CommandKey:%s, URI:%s, 请求参数:%s, 请求头:%s, HTTP状态码:%d, " +
                        "返回参数:%s, 耗时:%dms",
                invocation.getSoaRequestMethod(), invocation.getCommandKey(), uri, requestParams,
                requestHeaders, simpleHttpResponse.getCode(), responseBody, requestDuration);
        logger.debug(msg);
    }


    @Override
    public ServiceInstance selectServiceInstance(SoaRef soaRef) {
        List<ServiceInstance> serviceInstanceList = serviceRegistryDiscovery.discover(soaRef.value());
        if (serviceInstanceList.isEmpty()) {
            throw new RpcException(RpcException.NO_INSTANCE_AVAILABLE, "No service instance found for " + soaRef.value());
        }
        LoadBalance loadBalance = soaRef.loadbalance();
        ServiceInstance serviceInstance = serviceRegistryDiscovery.loadbalanceSelect(serviceInstanceList, loadBalance);
        return serviceInstance;
    }

    @Override
    public Invocation buildInvocation(Method proxyMethod, SoaRef soaRef, Object[] args) throws RpcException {
        try {
            return doBuildInvocation(proxyMethod, soaRef, args);
        } catch (Exception e) {
            throw new RpcException(RpcException.SOA_DEFINITION, "Build invocation failed", e);
        }
    }

    private RpcInvocation doBuildInvocation(Method proxyMethod, SoaRef soaRef, Object[] args) {
        Class<?> returnType = proxyMethod.getReturnType();
        Type type = proxyMethod.getGenericReturnType();
        Type[] types = TypeUtils.getReturnTypes(returnType, type);
        SoaMethod soaMethod = proxyMethod.getAnnotation(SoaMethod.class);
        Class<?> fallback = soaRef.fallback();
        String commandKey = buildCommandKey(proxyMethod, soaRef);
        Protocol protocol = parseInvokeType(soaMethod);
        RpcInvocation rpcInvocation = new RpcInvocation(protocol);
        rpcInvocation.put(SoaConstants.CONSUMER_MODEL, MetaData.getConsumerMeta(commandKey));
        rpcInvocation.putRawAttachment(SoaConstants.UP_STREAM_APP_ID, Constants.getAppid());
        rpcInvocation.setRemoteApplication(soaRef.value());
        rpcInvocation.setReturnType(returnType);
        String methodName = proxyMethod.getName();
        SoaRequestMethod soaRequestMethod = SoaRequestMethod.POST;
        SOAParamsMode soaParamsMode = SOAParamsMode.STANDARD;
        SOAResultMode soaResultMode = SOAResultMode.OBJECT;
        int readTimeout = 0;
        int connectionTimeout = 0;
        ServiceInstance targetService = null;
        String url = null;
        if (!Objects.isNull(soaMethod)) {
            rpcInvocation.setMethodName(soaMethod.methodName());
            rpcInvocation.setMethodPath(soaMethod.path());
            if (soaMethod.fallback() != Void.class) {
                fallback = soaMethod.fallback();
            }
            url = soaRef.url();
            if (!url.isEmpty()) {
                targetService = extractFromUrl(soaRef);
            }
            methodName = soaMethod.methodName();
            soaRequestMethod = soaMethod.http().method();
            readTimeout = soaMethod.timeout();
            soaParamsMode = soaMethod.http().paramsMode();
            soaResultMode = soaMethod.http().resultMode();
            connectionTimeout = soaMethod.connectionTimeout();
        }
        if (StringUtils.isEmpty(url)) {
            // 统计 服务发现时间
            long startTime = System.currentTimeMillis();
            targetService = selectServiceInstance(soaRef);
            long endTime = System.currentTimeMillis();
            long serviceDiscoveryDuration = endTime - startTime;
        }
        if (soaParamsMode.equals(SOAParamsMode.STANDARD)) {
            rpcInvocation.setProtocol(soaRef.protocol());
        }
        rpcInvocation.setHost(targetService.getHost());
        rpcInvocation.setPort(targetService.getPort());
        rpcInvocation.setHttps(targetService.isSecure());
        rpcInvocation.setCommandKey(commandKey);
        rpcInvocation.setArguments(args);
        rpcInvocation.setFallback(fallback);
        rpcInvocation.setServicePath(soaRef.path());
        rpcInvocation.setMethodName(methodName);
        Assert.notNull(soaRequestMethod, "SoaRequestMethod is required");
        rpcInvocation.setSoaRequestMethod(soaRequestMethod);
        rpcInvocation.setTimeout(readTimeout);
        rpcInvocation.setHttps(soaRef.https());
        rpcInvocation.setReturnType(returnType);
        rpcInvocation.setReturnTypes(types);
        rpcInvocation.setParamsMode(soaParamsMode);
        rpcInvocation.setResultMode(soaResultMode);
        rpcInvocation.setConnectionTimeout(connectionTimeout);
        return rpcInvocation;
    }


    private static String buildCommandKey(Method method, SoaRef soaRef) {
        String commandKey = soaRef.value() + "@" + soaRef.path() + "/" + method.getName();
        return commandKey;
    }

    /**
     * 从URL中提取服务实例信息,针对非注册中心的URL
     *
     * @param soaRef
     * @return
     */
    private ServiceInstance extractFromUrl(SoaRef soaRef) {
        String urlString = soaRef.url();
        if (urlString == null || urlString.trim().isEmpty()) {
            throw new RpcException(RpcException.SOA_DEFINITION);
        }
        URI uri = null;
        try {
            urlString = environment.resolvePlaceholders(urlString);
            uri = new URI(urlString);
        } catch (URISyntaxException e) {
            throw new RpcException(RpcException.SOA_DEFINITION);
        }
        String schema = uri.getScheme();
        if (!schema.startsWith("http")) {
            throw new RpcException(RpcException.SOA_DEFINITION);
        }
        String host = uri.getHost();
        int port = uri.getPort();
        ServiceInstance serviceInstance = new ServiceInstance(soaRef.value(), host, port);
        if (schema.equals("http")) {
            serviceInstance.setSecure(false);
        }
        return serviceInstance;

    }


    /**
     * 获取调用协议
     *
     * @param soaMethod
     * @return
     */
    private Protocol parseInvokeType(SoaMethod soaMethod) {
        if (soaMethod == null) {
            return Protocol.JSON_RPC;
        }
        if (soaMethod.http().paramsMode().equals(SOAParamsMode.STANDARD)) {
            return Protocol.JSON_RPC;
        }
        return Protocol.REST;
    }

    public Object handleFallback(Invocation invocation) {
        Class<?> fallbackClass = invocation.getFallback();

        if (fallbackClass == Void.class) {
            logger.warn("No fallback configured for method: {}", invocation.getMethodName());
            return null; // 或者抛出异常，根据需求决定
        }

        try {
            Object fallbackInstance = applicationContext.getBean(fallbackClass);
            if (fallbackInstance instanceof Fallback) {
                Fallback fallback = (Fallback) fallbackInstance;
                return fallback.handle(invocation);
            }
            throw new SoaFallbackException("Fallback class must implement Fallback interface");
        } catch (Exception e) {
            logger.error("Fallback failed", e);
            throw new SoaFallbackException("Fallback failed", e);
        }
    }
}